package com.llkj.common.http;

import com.llkj.common.exception.ApiException;
import com.llkj.common.model.BaseResponse;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * author: zhu.px
 * e-mail: 1271901673@qq.com
 * time  : 2017/6/13
 * desc  : 转换器
 */
public class RxHelper {

    public static <T> ObservableTransformer<BaseResponse<T>, T> applySchedulers() {
        return new ObservableTransformer<BaseResponse<T>, T>() {
            @Override
            public ObservableSource<T> apply(@NonNull Observable<BaseResponse<T>> upstream) {
                return upstream
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .retryWhen(new RetryWhenFun(HttpUtils.getInstance().getRetryCount(), HttpUtils.getInstance().getRetryDelayMillis()))
                        .flatMap(new Function<BaseResponse<T>, ObservableSource<T>>() {
                            @Override
                            public ObservableSource<T> apply(@NonNull BaseResponse<T> tHttpResult) {
                                if (tHttpResult.isOk()) {
                                    return createData(tHttpResult.getData());
                                } else {
                                    return Observable.error(new ApiException(tHttpResult.getCode(), tHttpResult.getMsg()));
                                }
                            }
                        });
            }
        };
    }

    private static <T> Observable<T> createData(final T t) {
        return Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<T> e) {
                if (!e.isDisposed()) {
                    try {
                        e.onNext(t);
                        e.onComplete();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                        e.onError(ex);
                    }
                }
            }
        });
    }

    public static <T> ObservableTransformer<T, T> io_main() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
                return upstream
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

}
